package com.mysite.seckill.message.task;

import java.util.Date;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSON;
import com.mysite.seckill.commons.common.Constant;
import com.mysite.seckill.commons.po.SeckillRecord;
import com.mysite.seckill.message.mapper.SeckillRecordMapper;

@Component
public class SeckillHandlerTask {

    protected final Logger logger = LoggerFactory.getLogger(this.getClass());

    @Autowired
    private SeckillRecordMapper seckillRecordMapper;

    @KafkaListener(topics = Constant.KAFKA_SECKILL_TOPIC, groupId = Constant.KAFKA_SECKILL_GROUP)
    public void listen(ConsumerRecord<String, String> record) {
        String value = record.value();
        System.out.println(value);
        System.out.println(record);

        SeckillRecord seckillRecord = JSON.parseObject(value, SeckillRecord.class);
        seckillRecord.setSeckillTime(new Date());
        System.err.println(seckillRecord);
        seckillRecordMapper.insertSeckillRecord(seckillRecord);
    }
}
